package com.baishancloud.log.test

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.entity._
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.source.{BroadcastUtil, SourceUtil}
import com.baishancloud.log.common.util.ExternalDataFetchUtil
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable
import scala.util.control.Breaks

/**
 * 测试获取所有资源的广播
 *
 * @author ziqiang.wang
 * @date 2022/2/18 14:28
 */
object TestCacheData2 extends Serializable {

  def main(args: Array[String]): Unit = {
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println(s"输入参数：${parameterTool.toMap.toString}")
    StreamEnv.builder(parameterTool)
    val source: DataStream[Long] = SourceUtil.numberEmit()
    val customInfoFetch: ExternalDataEntity = ExternalDataEntity
      .Builder()
      .enableDomainInfoFetch()
      .enableAccountDomainInfo("mfniu,adesk_20200629_del")
      .enableServerCacheGroupInfo()
      .enableCountryInfoFetch()
      .enableProvinceInfoFetch()
      .enableIspInfoFetch()
      .enableNodeInfoFetch()
      .enableServiceTypeInfoFetch()
      .enableTypeModuleInfoFetch()
      .enableSvgInfoFetch()
      .build()
    val info: BroadcastStream[ExternalDataEntity] = BroadcastUtil.customInfo(customInfoFetch)

    source
      .connect(info)
      .process(new BroadcastProcessFunction[Long, ExternalDataEntity, Long] {
        var domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
        var accountDomainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
        var serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()
        var countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
        var provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
        var ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()
        var userInfo: mutable.Map[String, Int] = mutable.Map[String, Int]()
        var nodeInfo: mutable.Map[String, NodeEntity] = mutable.Map[String, NodeEntity]()
        var typeModuleInfo: mutable.Map[String, mutable.Set[Int]] = mutable.Map[String, mutable.Set[Int]]()
        var serviceTypeInfo: mutable.Map[String, Byte] = mutable.Map[String, Byte]()
        var svgInfo: mutable.Map[String, String] = mutable.Map[String, String]()

        val LOGGER: Logger = LoggerFactory.getLogger("info")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.domainInfoFetchFromFile(domainInfo)
          ExternalDataFetchUtil.accountDomainInfoFetchFromFile("mfniu,adesk_20200629_del", accountDomainInfo)
          ExternalDataFetchUtil.serverCacheGroupFetchFromFile(serverCacheGroupInfo)
          ExternalDataFetchUtil.countryInfoFetchFromFile(countryInfo)
          ExternalDataFetchUtil.provinceInfoFetchFromFile(provinceInfo)
          ExternalDataFetchUtil.ispInfoFetchFromFile(ispInfo)
          ExternalDataFetchUtil.userInfoFetchFromFile(userInfo)
          ExternalDataFetchUtil.nodeInfoFetchFromFile(nodeInfo)
          ExternalDataFetchUtil.typeModuleInfoFetchFromFile(typeModuleInfo)
          ExternalDataFetchUtil.serviceTypeInfoFetchFromFile(serviceTypeInfo)
          ExternalDataFetchUtil.svgInfoFetchFromFile(svgInfo)
          LOGGER.info(s"并行度编号：${subtask}，初始化完成\n" +
            s"domainInfo数据量：${domainInfo.size}，部分内容：\n${outStrDomainInfo()}\n" +
            s"accountDomainInfo数据量：${accountDomainInfo.size}，部分内容：\n${outStrAccountDomainInfo()}\n" +
            s"serverCacheGroupInfo数据量：${serverCacheGroupInfo.size}，部分内容：\n${outStrServerCacheGroupInfo()}\n" +
            s"countryInfo数据量：${countryInfo.size}，部分内容：\n${outStrCountryInfo()}\n" +
            s"provinceInfo数据量：${provinceInfo.size}，部分内容：\n${outStrProvinceInfo()}\n" +
            s"ispInfo数据量：${ispInfo.size}，部分内容：\n${outStrIspInfo()}\n" +
            s"userInfo数据量：${userInfo.size}，部分内容：\n${outStrUserInfo()}\n" +
            s"nodeInfo数据量：${nodeInfo.size}，部分内容：\n${outStrNodeInfo()}\n" +
            s"typeModuleInfo数据量：${typeModuleInfo.size}，部分内容：\n${outStrTypeModuleInfo()}\n" +
            s"serviceTypeInfo数据量：${serviceTypeInfo.size}，部分内容：\n${outStrServiceTypeInfo()}\n" +
            s"svgInfo数据量：${svgInfo.size}，部分内容：\n${outStrSvgInfo()}\n"
          )
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, ExternalDataEntity, Long]#ReadOnlyContext, out: Collector[Long]): Unit = {

        }

        override def processBroadcastElement(value: ExternalDataEntity, ctx: BroadcastProcessFunction[Long, ExternalDataEntity, Long]#Context, out: Collector[Long]): Unit = {
          if (value.domainInfo.nonEmpty) {
            domainInfo = value.domainInfo
            LOGGER.info(s"domainInfo${subtask}接收到广播数据，domainInfo表size为：${domainInfo.size}，部分内容：\n${outStrDomainInfo()}\n")
          }
          if (value.accountDomainInfo.nonEmpty) {
            accountDomainInfo = value.accountDomainInfo
            LOGGER.info(s"accountDomainInfo${subtask}接收到广播数据，accountDomainInfo表size为：${accountDomainInfo.size}，部分内容：\n${outStrAccountDomainInfo()}\n")
          }
          if (value.serverCacheGroupInfo.nonEmpty) {
            serverCacheGroupInfo = value.serverCacheGroupInfo
            LOGGER.info(s"serverCacheGroupInfo${subtask}接收到广播数据，serverCacheGroupInfo表size为：${serverCacheGroupInfo.size}，部分内容：\n${outStrServerCacheGroupInfo()}\n")
          }
          if (value.countryInfo.nonEmpty) {
            countryInfo = value.countryInfo
            LOGGER.info(s"countryInfo${subtask}接收到广播数据，countryInfo表size为：${countryInfo.size}，部分内容：\n${outStrCountryInfo()}\n")
          }
          if (value.provinceInfo.nonEmpty) {
            provinceInfo = value.provinceInfo
            LOGGER.info(s"provinceInfo${subtask}接收到广播数据，provinceInfo表size为：${provinceInfo.size}，部分内容：\n${outStrProvinceInfo()}\n")
          }
          if (value.ispInfo.nonEmpty) {
            ispInfo = value.ispInfo
            LOGGER.info(s"ispInfo${subtask}接收到广播数据，ispInfo表size为：${ispInfo.size}，部分内容：\n${outStrIspInfo()}\n")
          }
          if (value.userInfo.nonEmpty) {
            userInfo = value.userInfo
            LOGGER.info(s"userInfo${subtask}接收到广播数据，userInfo表size为：${userInfo.size}，部分内容：\n${outStrUserInfo()}\n")
          }
          if (value.nodeInfo.nonEmpty) {
            nodeInfo = value.nodeInfo
            LOGGER.info(s"nodeInfo${subtask}接收到广播数据，nodeInfo表size为：${nodeInfo.size}，部分内容：\n${outStrNodeInfo()}\n")
          }
          if (value.typeModuleInfo.nonEmpty) {
            typeModuleInfo = value.typeModuleInfo
            LOGGER.info(s"typeModuleInfo${subtask}接收到广播数据，typeModuleInfo表size为：${typeModuleInfo.size}，部分内容：\n${outStrTypeModuleInfo()}\n")
          }
          if (value.serviceTypeInfo.nonEmpty) {
            serviceTypeInfo = value.serviceTypeInfo
            LOGGER.info(s"serviceTypeInfo${subtask}接收到广播数据，serviceTypeInfo表size为：${serviceTypeInfo.size}，部分内容：\n${outStrServiceTypeInfo()}\n")
          }
          if (value.svgInfo.nonEmpty) {
            svgInfo = value.svgInfo
            LOGGER.info(s"svgInfo${subtask}接收到广播数据，svgInfo表size为：${svgInfo.size}，部分内容：\n${outStrSvgInfo()}\n")
          }
        }

        def outStrDomainInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(domainInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            domainInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.toString()
        }

        def outStrAccountDomainInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(accountDomainInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            accountDomainInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrServerCacheGroupInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(serverCacheGroupInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            serverCacheGroupInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrCountryInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(countryInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            countryInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrProvinceInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(provinceInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            provinceInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrIspInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(ispInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            ispInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrUserInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(userInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            userInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(tuple._2).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrNodeInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(nodeInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            nodeInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrTypeModuleInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(typeModuleInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            typeModuleInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(JSONUtil.toJsonStr(tuple._2.toArray)).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrServiceTypeInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(serviceTypeInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            serviceTypeInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(tuple._2.toString).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

        def outStrSvgInfo(): String = {
          val builder = new StringBuilder()
          val max: Int = Math.min(svgInfo.size, 5)
          var count: Int = 0
          Breaks.breakable {
            svgInfo.foreach(tuple => {
              builder.append(tuple._1).append("->").append(tuple._2).append("\n")
              count += 1
              if (count >= max) Breaks.break()
            })
          }
          builder.deleteCharAt(builder.length - 1).toString()
        }

      })
      .print()


    StreamEnv.execute(parameterTool)

  }


}
